{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Chicago Crime Prediction Pipeline\n",
    "\n",
    "An example notebook that demonstrates how to:\n",
    "* Download data from BigQuery\n",
    "* Create a Kubeflow pipeline\n",
    "* Include Google Cloud AI Platform components to train and deploy the model in the pipeline\n",
    "* Submit a job for execution\n",
    "* Query the final deployed model\n",
    "\n",
    "The model forecasts how many crimes are expected to be reported the next day, based on how many were reported over the previous `n` days."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%capture\n",
    "\n",
    "# Install the SDK (Uncomment the code if the SDK is not installed before)\n",
    "!python3 -m pip install 'kfp>=0.1.31' --quiet\n",
    "!python3 -m pip install pandas --upgrade -q\n",
    "\n",
    "# Restart the kernel for changes to take effect"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import json\n",
    "\n",
    "import kfp\n",
    "import kfp.components as comp\n",
    "import kfp.dsl as dsl\n",
    "\n",
    "import pandas as pd\n",
    "\n",
    "import time"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Constants"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "tags": [
     "parameters"
    ]
   },
   "outputs": [],
   "source": [
    "# Required Parameters\n",
    "project_id = '<ADD GCP PROJECT HERE>'\n",
    "output = 'gs://<ADD STORAGE LOCATION HERE>' # No ending slash\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Optional Parameters\n",
    "REGION = 'us-central1'\n",
    "RUNTIME_VERSION = '1.13'\n",
    "PACKAGE_URIS=json.dumps(['gs://chicago-crime/chicago_crime_trainer-0.0.tar.gz'])\n",
    "TRAINER_OUTPUT_GCS_PATH = output + '/train/output/' + str(int(time.time())) + '/'\n",
    "DATA_GCS_PATH = output + '/reports.csv'\n",
    "PYTHON_MODULE = 'trainer.task'\n",
    "PIPELINE_NAME = 'Chicago Crime Prediction'\n",
    "PIPELINE_FILENAME_PREFIX = 'chicago'\n",
    "PIPELINE_DESCRIPTION = ''\n",
    "MODEL_NAME = 'chicago_pipeline_model' + str(int(time.time()))\n",
    "MODEL_VERSION = 'chicago_pipeline_model_v1' + str(int(time.time()))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Download data\n",
    "\n",
    "Define a download function that uses the BigQuery component"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "bigquery_query_op = comp.load_component_from_url(\n",
    "    'https://raw.githubusercontent.com/kubeflow/pipelines/01a23ae8672d3b18e88adf3036071496aca3552d/components/gcp/bigquery/query/component.yaml')\n",
    "\n",
    "QUERY = \"\"\"\n",
    "    SELECT count(*) as count, TIMESTAMP_TRUNC(date, DAY) as day\n",
    "    FROM `bigquery-public-data.chicago_crime.crime`\n",
    "    GROUP BY day\n",
    "    ORDER BY day\n",
    "\"\"\"\n",
    "\n",
    "def download(project_id, data_gcs_path):\n",
    "\n",
    "    return bigquery_query_op(\n",
    "        query=QUERY,\n",
    "        project_id=project_id,\n",
    "        output_gcs_path=data_gcs_path\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Train the model\n",
    "\n",
    "Run training code that will pre-process the data and then submit a training job to the AI Platform."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "mlengine_train_op = comp.load_component_from_url(\n",
    "    'https://raw.githubusercontent.com/kubeflow/pipelines/01a23ae8672d3b18e88adf3036071496aca3552d/components/gcp/ml_engine/train/component.yaml')\n",
    "\n",
    "def train(project_id,\n",
    "          trainer_args,\n",
    "          package_uris,\n",
    "          trainer_output_gcs_path,\n",
    "          gcs_working_dir,\n",
    "          region,\n",
    "          python_module,\n",
    "          runtime_version):\n",
    "\n",
    "    return mlengine_train_op(\n",
    "        project_id=project_id, \n",
    "        python_module=python_module,\n",
    "        package_uris=package_uris,\n",
    "        region=region,\n",
    "        args=trainer_args,\n",
    "        job_dir=trainer_output_gcs_path,\n",
    "        runtime_version=runtime_version\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Deploy model\n",
    "\n",
    "Deploy the model with the ID given from the training step"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "mlengine_deploy_op = comp.load_component_from_url(\n",
    "    'https://raw.githubusercontent.com/kubeflow/pipelines/01a23ae8672d3b18e88adf3036071496aca3552d/components/gcp/ml_engine/deploy/component.yaml')\n",
    "\n",
    "def deploy(\n",
    "    project_id,\n",
    "    model_uri,\n",
    "    model_id,\n",
    "    model_version,\n",
    "    runtime_version):\n",
    "    \n",
    "    return mlengine_deploy_op(\n",
    "        model_uri=model_uri,\n",
    "        project_id=project_id, \n",
    "        model_id=model_id, \n",
    "        version_id=model_version, \n",
    "        runtime_version=runtime_version, \n",
    "        replace_existing_version=True, \n",
    "        set_default=True)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Define pipeline"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@dsl.pipeline(\n",
    "    name=PIPELINE_NAME,\n",
    "    description=PIPELINE_DESCRIPTION\n",
    ")\n",
    "\n",
    "def pipeline(\n",
    "    data_gcs_path=DATA_GCS_PATH,\n",
    "    gcs_working_dir=output,\n",
    "    project_id=project_id,\n",
    "    python_module=PYTHON_MODULE,\n",
    "    region=REGION,\n",
    "    runtime_version=RUNTIME_VERSION,\n",
    "    package_uris=PACKAGE_URIS,\n",
    "    trainer_output_gcs_path=TRAINER_OUTPUT_GCS_PATH,\n",
    "):      \n",
    "    download_task = download(project_id,\n",
    "                             data_gcs_path)\n",
    "\n",
    "    train_task = train(project_id,\n",
    "                       json.dumps(\n",
    "                           ['--data-file-url',\n",
    "                            '%s' % download_task.outputs['output_gcs_path'],\n",
    "                            '--job-dir',\n",
    "                            output]\n",
    "                       ),\n",
    "                       package_uris,\n",
    "                       trainer_output_gcs_path,\n",
    "                       gcs_working_dir,\n",
    "                       region,\n",
    "                       python_module,\n",
    "                       runtime_version)\n",
    "    \n",
    "    deploy_task = deploy(project_id,\n",
    "                         train_task.outputs['job_dir'],\n",
    "                         MODEL_NAME,\n",
    "                         MODEL_VERSION,\n",
    "                         runtime_version)    \n",
    "    return True\n",
    "\n",
    "# Reference for invocation later\n",
    "pipeline_func = pipeline"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Submit the pipeline for execution"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pipeline = kfp.Client().create_run_from_pipeline_func(pipeline, arguments={})\n",
    "\n",
    "# Run the pipeline on a separate Kubeflow Cluster instead\n",
    "# (use if your notebook is not running in Kubeflow - e.x. if using AI Platform Notebooks)\n",
    "# pipeline = kfp.Client(host='<ADD KFP ENDPOINT HERE>').create_run_from_pipeline_func(pipeline, arguments={})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Wait for the pipeline to finish"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "run_detail = pipeline.wait_for_run_completion(timeout=1800)\n",
    "print(run_detail.run.status)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Use the deployed model to predict (online prediction)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import os\n",
    "os.environ['MODEL_NAME'] = MODEL_NAME\n",
    "os.environ['MODEL_VERSION'] = MODEL_VERSION"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Create normalized input representing 14 days prior to prediction day."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%writefile test.json\n",
    "{\"lstm_input\": [[-1.24344569, -0.71910112, -0.86641698, -0.91635456, -1.04868914, -1.01373283, -0.7690387, -0.71910112, -0.86641698, -0.91635456, -1.04868914, -1.01373283, -0.7690387 , -0.90387016]]}"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "!gcloud ai-platform predict --model=$MODEL_NAME --version=$MODEL_VERSION --json-instances=test.json"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Examine cloud services invoked by the pipeline\n",
    "- BigQuery query: https://console.cloud.google.com/bigquery?page=queries (click on 'Project History')\n",
    "- AI Platform training job: https://console.cloud.google.com/ai-platform/jobs\n",
    "- AI Platform model serving: https://console.cloud.google.com/ai-platform/models\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Clean models"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# !gcloud ai-platform versions delete $MODEL_VERSION --model $MODEL_NAME\n",
    "# !gcloud ai-platform models delete $MODEL_NAME"
   ]
  }
 ],
 "metadata": {
  "celltoolbar": "Tags",
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.7"
  },
  "pycharm": {
   "stem_cell": {
    "cell_type": "raw",
    "metadata": {
     "collapsed": false
    },
    "source": []
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}